package com.zhouyu.examplerocketmq.productor;

import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;

/**
 * @Description:发送事务性消息
 */
@Service
@Log4j2
public class TransActionMessageProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送事务性消息
     */
    private void sendTransactionMsg() {
        HashMap<String, String> headears = new HashMap<>();
        headears.put(RocketMQHeaders.TRANSACTION_ID, "454wr4e4re4re5rf");
        Message<String> message = MessageBuilder.withPayload("我是周宇5").copyHeaders(headears).build();

        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction("group1", "topic-5", message, null);
        TransActionMessageProducer.log.info("发送事务性消息结果，transactionSendResult{}", transactionSendResult);
    }


    /**
     * 事务性消息回查机制
     */
    @RocketMQTransactionListener(txProducerGroup = "group1")
    class TransactionListener implements RocketMQLocalTransactionListener {
        /**
         * 执行本地事务，这个时候消息已经成功的到达broker
         *
         * @param message
         * @param o
         * @return 返回执行本地事务的状态，之后就会告知MQ修改状态
         */
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
            //处理事务
            String transactionId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
            //缓存事务消息

            //返回消息状态
            return null;
        }

        /**
         * 检查事务状态
         * 例如当本地事务成功，但是由于网络原因在修改mq中的事务消息状态异常，就会导致事务消息的状态不一致。
         * 这个时候rocketmq会定期通过checkLocalTransaction方法来检查此类消息的状态，从而修改mq的消息状态。
         * 所以如果要核心业务用Rocketmq解决分布式事务问题，建议选择同步刷盘模式。
         *
         * @param message
         * @return
         */
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
            //检查消息状态，这个消息应该会redis(事务id,消息体，超时时间)
            return null;
        }
    }
}
